package com.atguigu.dga.governance.service.impl;

import com.atguigu.dga.common.util.SpringBeanProvider;
import com.atguigu.dga.ds.bean.TDsTaskDefinition;
import com.atguigu.dga.ds.bean.TDsTaskInstance;
import com.atguigu.dga.ds.service.TDsTaskDefinitionService;
import com.atguigu.dga.ds.service.TDsTaskInstanceService;
import com.atguigu.dga.governance.assessor.Assessor;
import com.atguigu.dga.governance.bean.AssessParam;
import com.atguigu.dga.governance.bean.GovernanceAssessDetail;
import com.atguigu.dga.governance.bean.GovernanceMetric;
import com.atguigu.dga.governance.mapper.GovernanceAssessDetailMapper;
import com.atguigu.dga.governance.service.GovernanceAssessDetailService;
import com.atguigu.dga.governance.service.GovernanceMetricService;
import com.atguigu.dga.meta.bean.TableMetaInfo;
import com.atguigu.dga.meta.bean.TableMetaInfoExtra;
import com.atguigu.dga.meta.service.TableMetaInfoExtraService;
import com.atguigu.dga.meta.service.TableMetaInfoService;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.text.ParseException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 * <p>
 * 治理考评结果明细 服务实现类
 * </p>
 *
 * @author zhangchen
 * @since 2024-11-22
 */
@DS("dga")
@Service
public class GovernanceAssessDetailServiceImpl extends ServiceImpl<GovernanceAssessDetailMapper, GovernanceAssessDetail> implements GovernanceAssessDetailService {

    @Autowired
    TableMetaInfoService tableMetaInfoService;

    @Autowired
    TableMetaInfoExtraService tableMetaInfoExtraService;

    @Autowired
    GovernanceMetricService governanceMetricService;

    @Autowired
    SpringBeanProvider springBeanProvider;

    List <GovernanceAssessDetail> assessDetailList ;

    AtomicLong atomicLong;

    @Autowired
    TDsTaskInstanceService tDsTaskInstanceService;
    @Autowired
    TDsTaskDefinitionService tDsTaskDefinitionService;

    ThreadPoolExecutor poolExecutor=   new ThreadPoolExecutor(10, 20, 60000,TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(1000));


    public  void  assessMetrics(String assessDate) throws  Exception {

        // 1 查询所有要考评的表(带辅助信息)   List<TableMetaInfo>
        //循环查询数据库 不推荐
//        List<TableMetaInfo> tableMetaInfoList = tableMetaInfoService.list(new QueryWrapper<TableMetaInfo>().eq("assess_date", assessDate));
//        for (TableMetaInfo tableMetaInfo : tableMetaInfoList) {
//            TableMetaInfoExtra tableMetaInfoExtra = tableMetaInfoExtraService.getOne(new QueryWrapper<TableMetaInfoExtra>().eq("table_name", tableMetaInfo.getTableName));
//            tableMetaInfo.setTableMetaInfoExtra(tableMetaInfoExtra);
//        }


        List<TableMetaInfo> tableMetaInfoList =  tableMetaInfoService.getTableMetaInfoListWithExtra( assessDate);
        Map<String, TableMetaInfo> tableMetaInfoMap = tableMetaInfoList.stream()
                .collect(Collectors.toMap(tableMetaInfo -> tableMetaInfo.getSchemaName() + "." + tableMetaInfo.getTableName(), tableMetaInfo -> tableMetaInfo));

        // System.out.println("tableMetaInfoList = " + tableMetaInfoList);
        // 2 查询所有要考评的指标  List<GovernanceMetric>

        List<GovernanceMetric> metricList = governanceMetricService.list(new QueryWrapper<GovernanceMetric>().eq("is_disabled", "0"));

        //3 关于ds的信息
        //3.1  任务实例集合
        //assessDate +1 天
        Date date = DateUtils.parseDate(assessDate, "yyyy-MM-dd");
        Date dsSubmitDate = DateUtils.addDays(date, 1);
        String submitDateStr = DateFormatUtils.format(dsSubmitDate, "yyyy-MM-dd");

        List<TDsTaskInstance> taskInstanceList = tDsTaskInstanceService.list(new QueryWrapper<TDsTaskInstance>().eq("date_format(submit_time,'%Y-%m-%d')",submitDateStr)) ;
        //为了优化性能，能否把list转为hashmap   key: schema.tableName  value: List<TDsTaskInstance>
        Map<String, List<TDsTaskInstance>> taskInstanceMap =taskInstanceList.stream().collect(Collectors.groupingBy(taskInstance -> taskInstance.getName()));


        //3.2  任务定义
        QueryWrapper<TDsTaskDefinition> tDsTaskDefinitionQueryWrapper = new QueryWrapper<>();
        tDsTaskDefinitionQueryWrapper.inSql("code", " select  task_code from t_ds_task_instance ti where  date_format(submit_time,'%Y-%m-%d') ='"+submitDateStr+"'");

        List<TDsTaskDefinition> taskDefinitionList = tDsTaskDefinitionService.list(tDsTaskDefinitionQueryWrapper) ;
        tDsTaskDefinitionService.extractSqlFromTaskDefineList(taskDefinitionList);

        Map<String, TDsTaskDefinition> dsTaskDefinitionMap = taskDefinitionList.stream().collect(Collectors.toMap(taskDefinition -> taskDefinition.getName(), taskDefinition -> taskDefinition));


        //4  考评所有表的所有指标  List<GovernanceAssessDetail>
        long startMs = System.currentTimeMillis();

        List<GovernanceAssessDetail> assessDetailList = new ArrayList<>(tableMetaInfoList.size()*metricList.size());
        //准备期货列表
        List<CompletableFuture<GovernanceAssessDetail>> futureList= new ArrayList<>(tableMetaInfoList.size()*metricList.size());

        for (TableMetaInfo tableMetaInfo : tableMetaInfoList) {
            for (GovernanceMetric metric : metricList) {

                Assessor assessor = springBeanProvider.getBean(metric.getMetricCode(), Assessor.class);
                AssessParam assessParam = new AssessParam( );
                assessParam.setGovernanceMetric(metric);
                assessParam.setAssessDate(assessDate);
                assessParam.setTableMetaInfo(tableMetaInfo);
                assessParam.setTableMetaInfoList(tableMetaInfoList);
                assessParam.setTableMetaInfoMap(tableMetaInfoMap);



                // 从定义列表中找出对应表的定义
                TDsTaskDefinition tDsTaskDefinition = dsTaskDefinitionMap.get(tableMetaInfo.getSchemaName() + "." + tableMetaInfo.getTableName());
                 assessParam.setTDsTaskDefinition(tDsTaskDefinition);
                // 从实例列表中找出对应表的实例
                List<TDsTaskInstance> tDsTaskInstanceList = taskInstanceMap.get(tableMetaInfo.getSchemaName() + "." + tableMetaInfo.getTableName());
                assessParam.setTDsTaskInstanceList(tDsTaskInstanceList);

// 建造者模式
//                AssessParam assessParam1 = AssessParam.builder().tableMetaInfoList(tableMetaInfoList)
//                        .assessDate(assessDate)
//                        .tDsTaskDefinition(tDsTaskDefinition).build();

                CompletableFuture<GovernanceAssessDetail> future = CompletableFuture.supplyAsync(() -> {
                    GovernanceAssessDetail assessDetail = assessor.assess(assessParam);
                    return assessDetail;

                },poolExecutor);

                futureList.add(future);
            }
        }

         assessDetailList = futureList.stream().map(future -> future.join()).collect(Collectors.toList());

        long endMs = System.currentTimeMillis();

        System.out.println("(endMs-startMs) = " + (endMs-startMs)); //3秒左右


        //5  保存到数据库中
        remove(new QueryWrapper<GovernanceAssessDetail>().eq("assess_date",assessDate));
        saveBatch(assessDetailList);

    }

}
